package com.innodealing.service;

import com.innodealing.offsetskip.annotation.KafkaOffsetSkip;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import java.util.List;

/**
 * Kafka批量消息消费者示例
 * 演示如何使用@KafkaOffsetSkip注解处理批量消息
 */
@Service
public class SpringKafkaListenerExample {

    /**
     * 标准批量消费者 - 使用基础的批量配置
     */
    @KafkaListener(topics = "spring_kafka_batch_example", groupId = "spring-kafka-batch-example-group", containerFactory = "batchManualKafkaListenerContainerFactory")
    @KafkaOffsetSkip
    public void consumeBatch(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        System.out.println("[spring_kafka_batch_example] 收到批量消息，数量: " + records.size());

        try {
            // 批量处理消息
            for (ConsumerRecord<String, String> record : records) {
                String value = record.value();

                // 模拟业务处理逻辑
                if (value.contains("error")) {
                    throw new RuntimeException("批量处理异常: " + value);
                }

                System.out.println("[spring_kafka_batch_example] 处理消息: partition=" + record.partition() +
                        ", offset=" + record.offset() + ", value=" + value);
            }

            System.out.println("[spring_kafka_batch_example] 批量处理成功，消息数量: " + records.size());
            ack.acknowledge();

        } catch (Exception e) {
            System.err.println("[spring_kafka_batch_example] 批量消息处理失败: " + e.getMessage());
            throw e;
        }
    }

}
